/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.core.channel.impl;

import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.osgi.framework.ServiceRegistration;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.base.adaptor.AbstractRegisteredService;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.common.Task.Source;
import com.cqcis.uip.base.framework.core.Endpoint;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.RegisteredServiceRepository;
import com.cqcis.uip.base.framework.core.channel.Channel;
import com.cqcis.uip.base.framework.core.cluster.ClusterSupportSelector;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;
import com.cqcis.uip.base.framework.utils.EndpointGenerator;
import com.cqcis.uip.base.framework.utils.LoopThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.StringUtil;
import com.cqcis.uip.base.logadaptor.LogConfiguration;

/**
 * @author huangmy
 * @date   2008-1-10
 */
public class DefaultChannelImpl implements Channel {
	private static Log log = LogFactory.getLog(DefaultChannelImpl.class);
	private static String CHANNEL_THREAD_NAME = "uip-context-channel";
	private RegisteredServiceRepository serviceRepository; // 服务集
	private LoopThreadPoolExecutor executor; // 线程池
	private int threadNumber; // 线程数
	private RegisteredServiceContext serviceContext; /*上下文*/
	private LogConfiguration logConfiguration; // 日志服务
	private ConcurrentTaskQueue<Task> taskDispatchQueue = new ConcurrentTaskQueue<Task>();
	private ClusterSupportSelector clusterSupportSelector;
	private ServiceRegistration clusterSupportRegistration; // 命令注册
	
	public void start(RegisteredServiceContext serviceContext) throws Exception {
		this.serviceRepository = serviceContext.getServiceRepository();
		this.serviceContext = serviceContext;
		this.clusterSupportSelector = new ClusterSupportSelector();
		this.clusterSupportSelector.start(serviceContext);
		
		this.clusterSupportRegistration = serviceContext.getBundleContext().registerService(ClusterSupportSelector.class.getName(), 
				this.clusterSupportSelector, null);
		
		this.executor = LoopThreadPoolExecutor.getExecutor(threadNumber, threadNumber, 
				60, serviceContext.getLogService(), logConfiguration, CHANNEL_THREAD_NAME, 10 * 1000);			
	}

	public void stop(RegisteredServiceContext serviceContext) throws Exception {
	    if (this.executor != null) {
	    	this.executor.shutdown();
	    }
	    
	    if (this.clusterSupportSelector != null) {
	    	this.clusterSupportSelector.stop(serviceContext);
	    }
	    
	    if (this.clusterSupportRegistration != null) {
			this.clusterSupportRegistration.unregister();
		}
	}
	
	public boolean scheduleSync(Task task) throws Exception {
		// check route
		checkRouteForTask(task);
		
		return doProcessForTaskDispatch(task, true);
	}
		
	public boolean schedule(Task task) throws Exception {
		// check route
		checkRouteForTask(task);
		
		boolean b = taskDispatchQueue.offer(task);
		if (executor.getPoolSize() < taskDispatchQueue.size() && executor.getPoolSize() < threadNumber) {
			executor.execute(new Runnable() {
				public void run() {
					while (true) {
						if (!processTaskRun()) {
						    break;
						}
					}
				}
			});
		}
		
		executor.notifyPool();
		
		return b;
	}
	
	/**
	 * 处理任务分发
	 * 
	 */
	private boolean processTaskRun() {		
		Task task = taskDispatchQueue.poll(); // 阻塞取task
		if (task == null) {
			return false;
		}
		
		return doProcessForTaskDispatch(task, false);
	}
	
	/**
	 * 任务分发
	 * 
	 * @param task
	 * @return
	 */
	private boolean doProcessForTaskDispatch(Task task, boolean sync) {
		RegisteredService rs = null;
		task.setState(State.DEALING);

		Endpoint toEndpoint = task.route().to();		
		if (toEndpoint != null) {
			String location = toEndpoint.getAttribute(EndpointGenerator.ENDPOINT_EXPRESSION_LOCATION);			
			location = StringUtil.trimWhiteToNull(location);
			if (task.getSource() == Source.REMOTE
					|| location != null) {
				if (location == null) {
					task.setMessage("the host and localhost must not be null.");
				} else {
					// remote
					try {
						this.clusterSupportSelector.schedule(task);
						return true;
					} catch (Exception e) {
						task.setMessage(e.getMessage(), e);
					}
				}
			} else {
				rs = serviceRepository.getRegisteredService(toEndpoint);
				// log.debug("dispatch task " + task + " to : " + toEndpoint);
				if (rs != null) {
					boolean success = false;

					try {
						success = dispatch(task, rs, sync);
					} catch (Throwable e) {
						success = false;
					}

					if (success) {
						return true;
					}
				} else {
					task.setMessage(FrameworkConstants.TASK_STATE_MSG_SERVICE_DEACTIVE);
				}
			}
		} else {
			task.setMessage("task's to endpoint is null.");
		}
		
		if (task.getState() != State.ERROR) {
		    task.setState(State.ERROR);		
		}
		Endpoint fromEndpoint = task.route().from();
		
		if (fromEndpoint == null) {
			log.warn("the Task's fromEndpoint is null. task = " + task);
		}
		
		AbstractRegisteredService fromRs = (AbstractRegisteredService)serviceRepository.getRegisteredService(fromEndpoint);	
		if (fromRs != null) {
			fromRs.decrPendingSendTaskSize();
		}
		if (fromEndpoint != null && fromRs != null) {
			if (!sync) {
				fromRs.processResult(task); // 异步回调结果
			}
						
			fromRs.getTaskRecord().getSendTaskRepository().incrErrorPlatform(); // 目的服务接收的任务平台处理失败
		}
		
		if (rs != null) {
			rs.getTaskRecord().getRecvTaskRepository().incrErrorPlatform(); // 发送的任务平台处理失败
		}
		
		return false;
	}
	
	/**
	 * 分发任务
	 * 
	 * @param task
	 * @param rs
	 * @param sync
	 * @return
	 * @throws Exception
	 */
	private boolean dispatch(Task task, RegisteredService rs, boolean sync) throws Exception {
		List<LogConfiguration> oldLogConfigr = null;
				
		try {
			oldLogConfigr = serviceContext.getLogService().getLogConfiguration(Thread.currentThread());
			serviceContext.getLogService().unregister(Thread.currentThread());
			serviceContext.getLogService().register(rs.getLogConfiguration());
			
		    boolean success = sync ? rs.scheduleSync(task) : rs.schedule(task);			
		    if (!success) {
		    	if (task.getMessage() == null) {
		    		task.setMessage(FrameworkConstants.TASK_STATE_MSG_TASK_SCHEDULE_ERROR);
		    	}
		    	
				return false;
		    }
		} catch(Throwable e) {
			log.info("Task schedule Exception: ", e);
			if (task.getMessage() == null) {
	    		task.setMessage(FrameworkConstants.TASK_STATE_MSG_TASK_SCHEDULE_ERROR, e);
	    	}
			return false;
		} finally {
			if (oldLogConfigr != null) {
				serviceContext.getLogService().unregister();
				for (LogConfiguration config : oldLogConfigr) {
					serviceContext.getLogService().register(config);
				}				
			} else {
				serviceContext.getLogService().unregister();
			}
		}
		
		return true;
	}
	
	/**
	 * 检查task的路由设置
	 * 
	 * @param task
	 * @throws Exception
	 */
	private void checkRouteForTask(Task task) throws Exception {
		Endpoint toEndpoint = task.route().to();
		Endpoint fromEndpoint = task.route().from();
		
		if (toEndpoint == null) {
			throw new Exception("Route Setting Error: task's toEndpoint is null.");
		}
		
		if (fromEndpoint == null) {
			throw new Exception("Route Setting Error: task's fromEndpoint is null.");
		}
	}

	public void setThreadNumber(int threadNumber) {
		this.threadNumber = threadNumber;
	}

	public void setLogConfiguration(LogConfiguration logConfiguration) {
		this.logConfiguration = logConfiguration;
	}
}
